package com.atguigu;

import com.atguigu.common.Utils;
import com.atguigu.flink.Practice;
import com.atguigu.flink.Solution;
import com.atguigu.flink.source.impl.WordSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Runner {

    private static final Logger LOG = LoggerFactory.getLogger(Runner.class);
    public static void main(String[] args) {
        LOG.error("创建执行环境...");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> source = env.addSource(new WordSource());

        SingleOutputStreamOperator<Tuple2<String, Long>> solution = Solution.run(source);
        SingleOutputStreamOperator<Tuple2<String, Long>> practice = Practice.run(source);

        SingleOutputStreamOperator<Tuple2<String, Boolean>> streamOperator = solution.connect(practice).keyBy(0, 0)
                .flatMap(new RichCoFlatMapFunction<Tuple2<String, Long>, Tuple2<String, Long>, Tuple2<String, Boolean>>() {

                    @Override
                    public void flatMap1(Tuple2<String, Long> value, Collector<Tuple2<String, Boolean>> out) throws Exception {
                        Tuple2<String, Long> value2 = s2.value();
                        if (value2 != null) {
                            s2.clear();
                            out.collect(new Tuple2<>(value2.f0, value.f1.equals(value2.f1)));
                        } else {
                            s1.update(value);
                        }
                    }

                    @Override
                    public void flatMap2(Tuple2<String, Long> value, Collector<Tuple2<String, Boolean>> out) throws Exception {
                        Tuple2<String, Long> value2 = s1.value();
                        if (value2 != null) {
                            s1.clear();
                            out.collect(new Tuple2<>(value2.f0, value.f1.equals(value2.f1)));
                        } else {
                            s2.update(value);
                        }
                    }

                    private ValueState<Tuple2<String, Long>> s1;
                    private ValueState<Tuple2<String, Long>> s2;


                    @Override
                    public void open(Configuration parameters) throws Exception {
                        s1 = getRuntimeContext().getState(new ValueStateDescriptor<>("stream1", Types.TUPLE(Types.STRING, Types.LONG)));
                        s2 = getRuntimeContext().getState(new ValueStateDescriptor<>("stream2", Types.TUPLE(Types.STRING, Types.LONG)));
                    }

                });

        try {
            LOG.error("执行并收集结果...");
            long count = streamOperator.executeAndCollect(10000).stream().filter(x -> !x.f1).count();
            LOG.error("向服务器发送结果...");
            Utils.sendResult(count == 0);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
